package org.pz.controller;

import com.fasterxml.jackson.databind.json.JsonMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.pz.dto.DogDto;
import org.springframework.core.ParameterizedTypeReference;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import reactor.core.publisher.Flux;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.time.LocalTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 测试
 * */
@RestController
@RequestMapping("/test")
@Slf4j
public class TestController {

    ExecutorService executorService = Executors.newSingleThreadExecutor();

    /**
     * 测试流式接口，返回结果可以使用 ResponseEntity，也可以直接返回 SseEmitter，区别是 ResponseEntity 可以设置响应码等
     * */
    @GetMapping("/sse")
    public ResponseEntity<SseEmitter> sse() {

        //定义
        SseEmitter emitter = new SseEmitter();
        emitter.onTimeout(() -> System.out.println("timeout"));
        emitter.onCompletion(() -> System.out.println("completion"));
        emitter.onError((e) -> System.out.println("error"));

        //模拟数据
        List<DogDto> list = new ArrayList<>();
        list.add(new DogDto("pz", 30));
        list.add(new DogDto("yq", 30));
        list.add(new DogDto("yl", 28));
        list.add(new DogDto("xx", 26));

        executorService.submit(() -> {
            try {

                for(DogDto d : list) {
                    //emitter.send(SseEmitter.event().data(JSONObject.toJSONString(d)));
                    emitter.send(SseEmitter.event().name("test").id("id").data(new JsonMapper().writeValueAsString(d)));
                    Thread.sleep(1000);
                }
                emitter.complete();
            } catch (IOException | InterruptedException e) {
                emitter.completeWithError(e);
            }
        });

        return ResponseEntity.ok(emitter);
    }

    /**
     * 测试封装流式接口 SpringMvc + SseEmitter
     * */
    @GetMapping("/sse2")
    public SseEmitter sse2() {

        //超时时间
        SseEmitter emitter = new SseEmitter(60_000L); // 超时时间 60 秒

        //调用 sse 接口
        WebClient client = WebClient.create();
        client.get()
                .uri("http://localhost:9005/ai/test/sse")
                .accept(MediaType.TEXT_EVENT_STREAM)
                .retrieve()
                .bodyToFlux(ServerSentEvent.class) // 使用ServerSentEvent处理元数据
                .subscribe(
                        data ->{
                            try {
                                System.out.println("收到事件: " + data);
                                emitter.send(SseEmitter.event().data(data.data()));
                            }catch (IOException e){
                                emitter.completeWithError(e);
                            }
                        },  // onNext 处理每个数据项
                        error -> System.err.println("发生错误: " + error), // onError  处理错误
                        () -> {
                            System.out.println("事件流结束");
                            emitter.complete();
                        }   // onComplete   处理完成信号
                );
        // 处理客户端断开连接
        emitter.onCompletion(() -> {
            System.out.println("客户端断开连接");
            emitter.complete();
        });
        emitter.onTimeout(() -> System.out.println("连接超时"));

        return emitter;
    }

    /**
     * WebFlux + 响应式流
     * */
    @GetMapping(value = "/sse3", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<DogDto>> reactiveStream() {

        ParameterizedTypeReference<ServerSentEvent<DogDto>> type
                = new ParameterizedTypeReference<ServerSentEvent<DogDto>>() {};

        WebClient webClient = WebClient.create();
        Flux<ServerSentEvent<DogDto>> dogDtoFlux = webClient.get()
                .uri("http://localhost:9005/ai/test/sse")
                .accept(MediaType.TEXT_EVENT_STREAM)
                .retrieve()
                .bodyToFlux(type);

        dogDtoFlux.subscribe(
                content -> log.info("Time: {} - event: name[{}], id [{}], content[{}] ",
                        LocalTime.now(), content.event(), content.id(), content.data()),
                error -> log.error("Error receiving SSE: {}", error),
                () -> log.info("Completed!!!"));
        return dogDtoFlux; // 直接透传数据
    }

    /**
     * 通过流，手动解析响应流
     * */
    @GetMapping(value="/sse4", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public ResponseEntity<SseEmitter> see4() {

        //封装请求
        CloseableHttpClient httpClient = HttpClients.createDefault();
        HttpGet httpGet = new HttpGet("http://localhost:9005/ai/test/sse");

        //使用 SSeEmitter 返回信息
        SseEmitter emitter = new SseEmitter(60_000L);
        //ResponseEntity.BodyBuilder bodyBuilder = ResponseEntity.badRequest();

        executorService.submit( () -> {
        try (CloseableHttpResponse response = httpClient.execute(httpGet);
             BufferedReader reader = new BufferedReader(new InputStreamReader(response.getEntity().getContent()))) {

            //bodyBuilder = ResponseEntity.status(response.getStatusLine().getStatusCode());
            //bodyBuilder.header("response_msg", response.getStatusLine().toString());

            String line;
            SseEmitter.SseEventBuilder event = SseEmitter.event();
            while ((line = reader.readLine()) != null) {
                //空行表示伊特事件结束
                if (line.isEmpty()) {
                    //发送上一个事件
                    if (event != null) {
                        emitter.send(event);
                    }

                    //新建一个 event
                    event = SseEmitter.event();
                } else if (line.startsWith("event:")) {
                    event.name(line.substring(6).trim());
                } else if (line.startsWith("id:")) {
                    event.id(line.substring(3).trim());
                } else if (line.startsWith("data:")) {
                    event.data(line.substring(5).trim());
                }
            }
            emitter.complete();
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
        });

        return ResponseEntity.ok().body(emitter);
    }
}


